package org.ws.task.example.mq;

import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.ws.task.item.ItemLeaderProvider;
import org.ws.task.model.TaskItem;
import org.ws.task.model.protocol.PullItemRequest;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
public class FileLineItemProvider {

    private final int bufferSize = 1024 * 64;
    private final long fileLength;
    private final RandomAccessFile randomAccessFile;
    private AtomicLong startAtomic = new AtomicLong(0);
    private AtomicInteger count = new AtomicInteger(0);

    public FileLineItemProvider(String path) throws FileNotFoundException {
        File file = new File(path);
        if(!file.exists()){
            throw new FileNotFoundException(path+" File Not Found.");
        }
        this.fileLength = file.length();
        this.randomAccessFile = new RandomAccessFile(file,"r");
    }

    private long getEndPosition(long start) throws IOException {
        long endPosition = start+bufferSize-1;
        if(endPosition >= fileLength-1){
            return fileLength-1;
        }
        this.randomAccessFile.seek(endPosition);
        byte temp = this.randomAccessFile.readByte();
        // 截取到换行位置
        while (temp != '\n' && temp != '\r'){
            endPosition++;
            if(endPosition >= fileLength-1){
                return fileLength-1;
            }
            this.randomAccessFile.seek(endPosition);
            temp = this.randomAccessFile.readByte();
        }
        return endPosition;
    }

    private String[] read() throws IOException {
        long start = this.startAtomic.get();
        if(start>=fileLength-1){
            return null;
        }
        long endPosition = getEndPosition(start);
        long sliceSize = endPosition - start + 1;
        int length = ((Long)sliceSize).intValue();
        byte[] readBuff = new byte[length];
        MappedByteBuffer mappedByteBuffer = this.randomAccessFile
                .getChannel()
                .map(FileChannel.MapMode.READ_ONLY,start,sliceSize);
        mappedByteBuffer.get(readBuff,0,length);
        String lines = new String(readBuff,"UTF-8");
        this.startAtomic.addAndGet(sliceSize);
        return StringUtils.split(lines,"\n");
    }

    public Collection<TaskItem<String>> getItems() {
        List<TaskItem<String>> result = Lists.newArrayList();
        try {
            String[] lines = read();
            if(lines != null){
                for (String line : lines) {
                    result.add(new TaskItem<>(line));
                }
                log.info("result:{}",count.addAndGet(lines.length));
            }
        } catch (IOException e) {
            log.error(e.getMessage(),e);
        }
        return result;
    }
}
